package com.shuimutong.gmq.client.consumer;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;

import com.alibaba.fastjson.JSONObject;
import com.shuimutong.gmq.client.bean.HttpResponseBean;
import com.shuimutong.gmq.client.bean.ResponseDataVo;
import com.shuimutong.gmq.client.consumer.bean.MqContent;
import com.shuimutong.gmq.client.consumer.bean.MqDto;
import com.shuimutong.gmq.client.exception.SendMqException;
import com.shuimutong.gmq.client.util.HttpUtil;
import com.shuimutong.gmq.client.util.UrlPackUtil;
import com.shuimutong.gutil.common.GUtilCommonUtil;

/**
 * 消息消费方
 * @ClassName:  GmqProducer   
 * @Description:(这里用一句话描述这个类的作用)   
 * @author: 水木桶
 * @date:   2019年10月30日 下午9:37:58     
 * @Copyright: 2019 [水木桶]  All rights reserved.
 */
public class GmqConsumer {
	/**gmq服务的地址**/
	private String gmqServerUrl;
	/**gmq服务获取消息的全地址**/
	private String getMqUrl;

	public GmqConsumer(String gmqServerUrl) {
		this.gmqServerUrl = gmqServerUrl;
		this.getMqUrl = gmqServerUrl + UrlPackUtil.getGetMessagePath();
	}
	
	public List<MqContent> getMq(String topic, long offset, long size) throws SendMqException {
		List<MqContent> mqList = new ArrayList();
		Map<String, String> params = new HashMap();
		params.put("topic", topic);
		params.put("offset", String.valueOf(offset));
		params.put("size", String.valueOf(size));
		try {
			HttpResponseBean responseBean = HttpUtil.get(getMqUrl, params);
			boolean state = false;
			if(responseBean != null && StringUtils.isNotBlank(responseBean.getBody())) {
				ResponseDataVo responseData = JSONObject.parseObject(responseBean.getBody(), ResponseDataVo.class);
				if(responseData != null && responseData.getState() == 200) {
					state = true;
					if(responseData.getData() != null) {
						List<MqDto> tmpList = JSONObject.parseArray(responseData.getData().toString(), MqDto.class);
						if(!GUtilCommonUtil.checkListEmpty(tmpList)) {
							tmpList.forEach(mq -> {
								mqList.add(new MqContent(mq));
							});
						}
					}
				}
			}
		} catch (Exception e) {
			throw new SendMqException(e);
		}
		return mqList;
	}
}
